package org.etlt.extract;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.etlt.job.JobContext;
import org.etlt.job.JobSetting;
import org.etlt.job.SettingResolver;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.List;

public class ExtractorFactory {

    protected Log log = LogFactory.getLog(getClass());
    private static final ExtractorFactory EXTRACTOR_FACTORY = new ExtractorFactory();

    private ExtractorFactory() {
    }

    public static ExtractorFactory getInstance() {
        return EXTRACTOR_FACTORY;
    }

    private SettingResolver<ExtractorSetting> settingResolver;

    protected void populateSettingResolver(String resourceName) {
        if (JobSetting.isXml(resourceName))
            this.settingResolver = new ExtractorSettingXmlResolver();
        else throw new IllegalStateException("unsupported extract file: " + resourceName);
    }

//    public List<Extractor> resolveInstances(String[] settingFiles, JobContext context) throws IOException {
//        List<Extractor> extractors = new ArrayList<>();
//        for (String ext : settingFiles) {
//            List<Extractor> extractors1 =
//            extractors.add(resolveInstances(ext, context));
//        }
//        return extractors;
//    }

    public List<Extractor> resolveInstances(String settingFile, JobContext context) throws IOException {
        populateSettingResolver(settingFile);
        List<Extractor> results = new ArrayList<>();
        File file = new File(context.getContextRoot(), settingFile);
        log.info("start: resolve extractor(s) from file: " + file);
        InputStream stream = new FileInputStream(file);
        List<ExtractorSetting> extractorSettings =
                settingResolver.resolveBundle(stream, context);
        if (!extractorSettings.isEmpty()) {
            extractorSettings.forEach((setting) -> {
                results.add(createExtractor(setting, context));
            });
        } else {
            stream = new FileInputStream(new File(context.getContextRoot(), settingFile));
            results.add(createExtractor(settingResolver.resolve(stream, context), context));
        }
        log.info("end: resolve extractor(s) from file: " + file);
        return results;
    }

//    @Deprecated
//    public List<Extractor> resolveBundleInstances(String[] settingFiles, JobContext context) throws IOException {
//        List<Extractor> extractors = new ArrayList<>();
//        for (String ext : settingFiles) {
//            extractors.addAll(resolveBundleInstance(ext, context));
//        }
//        return extractors;
//    }

//    public List<Extractor> resolveBundleInstance(String settingFile, JobContext context) throws IOException {
//        BundleExtractorSetting extractorBundleSetting = this.reader.read(new File(context.getContextRoot(), settingFile), BundleExtractorSetting.class);
//        List<ExtractorSetting> extractorSettings = extractorBundleSetting.createExtractorSetting();
//        List<Extractor> extractors = new ArrayList<>(extractorSettings.size());
//        extractorSettings.forEach((setting) -> {
//            extractors.add(createExtractor(setting, context));
//        });
//        return extractors;
//    }

    public Extractor createExtractor(ExtractorSetting setting, JobContext context) {
        if (setting instanceof FileExtractSetting) {
            FileExtractor extractor = new FileExtractor((FileExtractSetting) setting);
            return extractor;
        } else if (setting instanceof DatabaseExtractSetting) {
            DatabaseExtractSetting databaseExtractSetting = (DatabaseExtractSetting) setting;
            DatabaseExtractor extractor;
            if (databaseExtractSetting.getPkDql() != null)
                extractor = new BigDatabaseExtractor(databaseExtractSetting);
            else
                extractor = new DatabaseExtractor(databaseExtractSetting);
            return extractor;
        } else if (setting instanceof LinkedExtractorSetting) {
            LinkedExtractor extractor = new LinkedExtractor((LinkedExtractorSetting) setting);
            return extractor;
        }
        throw new IllegalArgumentException("unsupported extractor setting: " + setting.getName());
    }
}
